package com.gitee.xhs;

import com.gitee.xhs.dao.WatermarkRecordDao;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/7/14 16:56
 * @description WatermarkRecordMakerProcessFunction
 */
public class WatermarkRecordMakerProcessFunction extends ProcessFunction<Transaction, WatermarkRecordDao<Transaction>> {

    private final String watermarkStrategyType;
    private final String watermarkGeneratorType;
    private final String watermarkAssignerType;

    public WatermarkRecordMakerProcessFunction(String watermarkStrategyType, String watermarkGeneratorType, String watermarkAssignerType) {
        this.watermarkStrategyType = watermarkStrategyType;
        this.watermarkGeneratorType = watermarkGeneratorType;
        this.watermarkAssignerType = watermarkAssignerType;
    }

    @Override
    public void processElement(Transaction value, ProcessFunction<Transaction, WatermarkRecordDao<Transaction>>.Context ctx, Collector<WatermarkRecordDao<Transaction>> out) throws Exception {
        WatermarkRecordDao<Transaction> watermarkRecordDao = new WatermarkRecordDao<>(watermarkStrategyType, watermarkGeneratorType, watermarkAssignerType);
        watermarkRecordDao.setRecord(value);
        watermarkRecordDao.setSystemCurrentTimeMillis(System.currentTimeMillis());
        watermarkRecordDao.setStreamRecordTimestamp(ctx.timestamp());
        watermarkRecordDao.setTimeServiceCurrentWatermark(ctx.timerService().currentWatermark());
        watermarkRecordDao.setTimeServiceProcessingTime(ctx.timerService().currentProcessingTime());
        out.collect(
                watermarkRecordDao
        );
    }


}
